import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.Executors;

public class AdvancedReactor {
    // 虚拟线程调度器
    static final Scheduler VIRTUAL_SCHEDULER =
            Schedulers.fromExecutor(Executors.newVirtualThreadPerTaskExecutor());

    public static Flux<String> reactiveProcessing() {
        return Flux.interval(Duration.ofMillis(100))
                .take(10)
                .publishOn(VIRTUAL_SCHEDULER)
                .map(i -> {
                    // 在虚拟线程中执行阻塞操作
                    return blockingOperation(i);
                });
    }

    private static String blockingOperation(long i) {
        try {
            // 模拟阻塞I/O
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "Processed-" + i;
    }
}